package com.weepal.pavement.mqtt.demo.message;

import com.weepal.pavement.mqtt.demo.config.mqtt.DefaultMqttProperties;
import com.weepal.pavement.mqtt.demo.config.mqtt.MqttClientContainer;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.stereotype.Component;

import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;

@Component
@Slf4j
public class MqttSender {
    @Autowired
    private MqttPahoClientFactory mqttFactory;
    @Autowired
    private MqttConnectOptions mqttConnectOptions;
    @Autowired
    private DefaultMqttProperties mqttProperties;

    public boolean sendMessage(String mqttId, String topic, byte[] message) {
        try {
            AtomicReference<IMqttClient> client = MqttClientContainer.getClient(mqttId);
            if (null == client || null == client.get()) {
                IMqttClient createClient = mqttFactory.getClientInstance(mqttProperties.getAddress(), mqttProperties.getClientId() + UUID.randomUUID().toString());
                createClient.setCallback(new PushCallback(mqttId));
                createClient.connect(mqttConnectOptions);
                AtomicReference<IMqttClient> mclient = MqttClientContainer.getClient(mqttId);
                if (mclient == null) {
                    mclient = new AtomicReference();
                }
                mclient.set(createClient);
                MqttClientContainer.putClient(mqttId, mclient);
                send(createClient, topic, message);
            } else {
                send(client.get(), topic, message);
            }
            return true;
        } catch (MqttException e) {
            AtomicReference<IMqttClient> client = MqttClientContainer.getClient(mqttId);
            if (!client.get().isConnected()) {
                MqttClientContainer.getClient(mqttId).set(null);
            }
            throw new RuntimeException(e);
        }
    }

    public void send(IMqttClient client, String topic, byte[] message) throws MqttException {
        MqttTopic t = client.getTopic(topic);
        MqttDeliveryToken token = t.publish(message, mqttProperties.getQos(), true);
        token.waitForCompletion();
        log.info("主题：" + topic + " 发送消息 : " + message);
    }
}
